package nsq

import (
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"github.com/nsqio/go-nsq"
)

type Consumer struct {
	Handler    nsq.Handler
	Topic      string
	Channel    string
	LookUpAddr string
}

func NewConsumer(handler nsq.Handler, addr, topic, channel string) *Consumer {
	return &Consumer{
		Handler:    handler,
		Topic:      topic,
		Channel:    channel,
		LookUpAddr: addr,
	}
}

func (c *Consumer) Run() {
	termChan := make(chan os.Signal, 1)
	signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM)

	consumer, err := nsq.NewConsumer(c.Topic, c.Channel, nsq.NewConfig())
	if err != nil {
		// TODO log error
		fmt.Println(err)
	}

	consumer.ChangeMaxInFlight(10)
	consumer.AddConcurrentHandlers(c.Handler, 5)
	// 建立NSQLookupd连接
	if err = consumer.ConnectToNSQLookupd(c.LookUpAddr); err != nil {
		// TODO log error
		fmt.Println(err)
	}

	for {
		select {
		case <-consumer.StopChan:
			return
		case <-termChan:
			consumer.Stop()
			return
		}
	}
}
